package com.gzx.qc.core.kafka;

import com.gzx.qc.modular.check.service.GzxCallCheckService;
import com.gzx.qc.modular.voice.text.service.GzxHtVoiceTextService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import vip.xiaonuo.common.kafka.KafkaTopicConstant;

import java.util.List;

/**
 * @author zjt
 * @description
 * @date 2025/2/18 15:39
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class QcEventKafkaConsumer {

    private final GzxCallCheckService callCheckService;

    private final GzxHtVoiceTextService htVoiceTextService;

    @KafkaListener(topics = KafkaTopicConstant.QC_CALL_RECEIPT_TOPIC, groupId = "my-consumer-group", containerFactory = "batchFactory")
    public void listen(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        log.info("从MQ中批量获取event：本批次处理event数量{}", recordList.size());
        try {
            callCheckService.saveCallCheck(recordList);
            ack.acknowledge();
        } catch (Exception e) {
            log.error("当前批次消息处理失败:{}", e.getMessage());
        }
    }

    @KafkaListener(topics = "voice-topic",
            groupId = "my-consumer-group", containerFactory = "batchFactory")
    public void saveVoiceText(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        log.info("从MQ中批量获取event：本批次处理event数量{}", recordList.size());
        try {
            htVoiceTextService.saveVoiceText(recordList);
        } catch (Exception e) {
            log.error("当前批次消息处理失败:{}", e.getMessage());
        }
        // 业务处理完毕再手动提交offset
        ack.acknowledge();
    }
}